#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/time.h>
#include <errno.h>
#include <netinet/in.h>
#include <unistd.h>
#include <sys/ioctl.h>

#include "conf.h"
#include "my_lock.h"
#include "global.h"
#include "my_memory.h"
#include "net/socket.h"
#include "log.h"
#include "net/netglobal.h"

// 构造函数
CSocket::CSocket()
{
    // 初始化配置相关（先给默认值，后续读配置）
    m_worker_connections = 1;      // epoll连接最大项数
    m_ListenPortCount = 1;         // 监听端口数
    m_RecyConnectionWaitTime = 60; // 回收等待时间

    // 初始化epoll相关
    m_epollhandle = -1; // epoll返回的句柄

    // 一些和网络通讯有关的成员变量（供后续频繁使用时提高效率）
    m_iLenMsgHeader = sizeof(STRUC_MSG_HEADER); // 消息头长度
    m_iLenPkgHeader = sizeof(COMM_PKG_HEADER);  // 包头长度

    // 互斥量的初始化交由子进程

    // 初始化各种队列相关
    m_iSendMsgQueueCount = 0;     // 发消息队列大小
    m_totol_recyconnection_n = 0; // 待释放队列大小
    m_cur_size_ = 0;              // 当前计时队列大小
    m_timer_value_ = 0;           // 当前计时队列头部的时间值
    m_iDiscardSendPkgCount = 0;   // 丢弃的发送数据包数量

    // 在线用户相关
    m_onlineUserCount = 0; // 在线用户数量统计
    m_lastprintTime = 0;   // 上次打印统计信息的时间
}

// 初始化函数（在创建子进程之前执行）
bool CSocket::Initialize()
{
    ReadConf(); // 读配置
    if (Open_listening_sockets() == false)
    {
        return false;
    }
    return true;
}

// 互斥量和线程初始化函数(初始化互斥量和线程)
bool CSocket::Initialize_subproc()
{
    // 发消息互斥量初始化
    if (pthread_mutex_init(&m_sendMessageQueueMutex, NULL) != 0)
    {
        Log_Error(0, "CSocket::Initialize_subproc()中pthread_mutex_init(&m_sendMessageQueueMutex)失败.");
        return false;
    }
    // 连接相关互斥量初始化
    if (pthread_mutex_init(&m_connectionMutex, NULL) != 0)
    {
        Log_Error(0, "CSocket::Initialize_subproc()中pthread_mutex_init(&m_connectionMutex)失败.");
        return false;
    }
    // 回收队列互斥量初始化
    if (pthread_mutex_init(&m_recyconnqueueMutex, NULL) != 0)
    {
        Log_Error(0, "CSocket::Initialize_subproc()中pthread_mutex_init(&m_recyconnqueueMutex)失败.");
        return false;
    }
    // 按时间排序的消息头处理队列互斥量初始化
    if (pthread_mutex_init(&m_timequeueMutex, NULL) != 0)
    {
        Log_Error(0, "CSocket::Initialize_subproc()中pthread_mutex_init(&m_timequeueMutex)失败.");
        return false;
    }

    // 初始化发消息相关的信号量（比条件变量方便）
    // 第二个参数0表示信号量在线程之间共享,非0表示在进程之间共享
    // 第三个参数0表示信号量初始值，为0时表示调用sem_wait()会进入阻塞
    if (sem_init(&m_semEventSendQueue, 0, 0) == -1)
    {
        Log_Error(0, "CSocket::Initialize_subproc()中sem_init(&m_semEventSendQueue,0,0)失败.");
        return false;
    }

    // 创建专门处理发送数据的线程
    int err;
    ThreadItem *pSendQueue; // 专门处理发送数据的线程
    m_threadVector.push_back(pSendQueue = new ThreadItem(this));
    err = pthread_create(&pSendQueue->_Handle, NULL, ServerSendQueueThread, pSendQueue);
    if (err != 0)
    {
        Log_Error(0, "CSocket::Initialize_subproc()中pthread_create(ServerSendQueueThread)失败.");
        return false;
    }

    // 创建专门用来回收连接的线程
    ThreadItem *pRecyconn; // 专门用来回收连接的线程
    m_threadVector.push_back(pRecyconn = new ThreadItem(this));
    err = pthread_create(&pRecyconn->_Handle, NULL, ServerRecyConnectionThread, pRecyconn);
    if (err != 0)
    {
        Log_Error(0, "CSocket::Initialize_subproc()中pthread_create(ServerRecyConnectionThread)失败.");
        return false;
    }

    // 根据是否开启踢人时钟来创建心跳包线程
    if (m_ifkickTimeCount == 1)
    {
        ThreadItem *pTimemonitor; // 专门用来处理到期不发心跳包的用户踢出的线程
        m_threadVector.push_back(pTimemonitor = new ThreadItem(this));
        err = pthread_create(&pTimemonitor->_Handle, NULL, ServerTimerQueueMonitorThread, pTimemonitor);
        if (err != 0)
        {
            Log_Error(0, "CSocket::Initialize_subproc()中pthread_create(ServerTimerQueueMonitorThread)失败.");
            return false;
        }
    }
    return true;
}

// 析构函数
CSocket::~CSocket()
{
    // 释放所有监听的端口相关内存
    std::vector<lplistening_t>::iterator pos;
    for (pos = m_ListenSocketList.begin(); pos != m_ListenSocketList.end(); ++pos)
    {
        delete (*pos); // 一定要把指针指向的内存干掉，不然内存泄漏
    }
    m_ListenSocketList.clear();
}

// 关闭退出函数(销毁线程)
void CSocket::Shutdown_subproc()
{
    // 把干活的线程停止掉，注意应该先尝试设置g_stopEvent = 1来开始让整个项目停止

    // 先尝试唤醒所有被信号量阻塞的线程
    if (sem_post(&m_semEventSendQueue) == -1)
    {
        Log_Error(0, "CSocket::Shutdown_subproc()中sem_post(&m_semEventSendQueue)失败.");
    }

    // 等待所有线程
    std::vector<ThreadItem *>::iterator iter;
    for (iter = m_threadVector.begin(); iter != m_threadVector.end(); iter++)
    {
        pthread_join((*iter)->_Handle, NULL); // 等待一个线程终止
    }

    // 遍历释放
    for (iter = m_threadVector.begin(); iter != m_threadVector.end(); iter++)
    {
        if (*iter)
        {
            delete (*iter);
        }
    }
    m_threadVector.clear();

    // 清理队列相关
    ClearMsgSendQueue();      // 清理发送消息队列
    Clearconnection();        // 清理连接池
    ClearAllFromTimerQueue(); // 清理心跳超时队列中所有内容

    // 销毁线程和信号量
    pthread_mutex_destroy(&m_connectionMutex);       // 连接相关互斥量释放
    pthread_mutex_destroy(&m_sendMessageQueueMutex); // 发消息互斥量释放
    pthread_mutex_destroy(&m_recyconnqueueMutex);    // 连接回收队列相关的互斥量释放
    pthread_mutex_destroy(&m_timequeueMutex);        // 心跳超时时间处理队列相关的互斥量释放
    sem_destroy(&m_semEventSendQueue);               // 发消息相关线程信号量释放
}

// 清理TCP发送消息队列函数
void CSocket::ClearMsgSendQueue()
{
    char *sTmpMempoint;
    CMemory *p_memory = CMemory::GetInstance();

    // 遍历释放内存
    while (!m_MsgSendQueue.empty())
    {
        sTmpMempoint = m_MsgSendQueue.front();
        m_MsgSendQueue.pop_front();
        p_memory->FreeMemory(sTmpMempoint);
    }
}

// 读取配置函数
void CSocket::ReadConf()
{
    CConfig *p_config = CConfig::GetInstance();
    // 读取epoll连接的最大项数
    m_worker_connections = p_config->GetIntDefault("net_worker_connections", m_worker_connections);
    // 取得要监听的端口数量
    m_ListenPortCount = p_config->GetIntDefault("net_ListenPortCount", m_ListenPortCount);
    // 读取回收等待时间
    m_RecyConnectionWaitTime = p_config->GetIntDefault("net_Sock_RecyConnectionWaitTime", m_RecyConnectionWaitTime);
    // 读取是否开启踢人时钟，1：开启   0：不开启
    m_ifkickTimeCount = p_config->GetIntDefault("net_Sock_WaitTimeEnable", 0);
    // 读取多少秒检测一次是否心跳超时
    m_iWaitTime = p_config->GetIntDefault("net_Sock_MaxWaitTime", m_iWaitTime);
    // 读取踢人策略（0直接踢，1忽视）
    m_ifTimeOutKick = p_config->GetIntDefault("net_Sock_TimeOutKick", 0);
    // 读取Flood攻击检测是否开启,1：开启   0：不开启
    m_floodAkEnable = p_config->GetIntDefault("net_Sock_FloodAttackKickEnable", 0);
    // 读取每次收到数据包的时间间隔(毫秒)
    m_floodTimeInterval = p_config->GetIntDefault("net_Sock_FloodTimeInterval", 100);
    // 读取累积多少次踢出此人
    m_floodKickCount = p_config->GetIntDefault("net_Sock_FloodKickCounter", 10);
}

// 开启端口监听
bool CSocket::Open_listening_sockets()
{
    int isock;                    // 暂存监听的socket套接字
    struct sockaddr_in serv_addr; // 暂存服务器的地址结构体
    int iport;                    // 暂存端口号

    // 进行初始化
    memset(&serv_addr, 0, sizeof(serv_addr));
    serv_addr.sin_family = AF_INET;                // 协议族为IPV4
    serv_addr.sin_addr.s_addr = htonl(INADDR_ANY); // 监听本地所有ip地址

    CConfig *p_config = CConfig::GetInstance();

    // 只监听一个端口即可
    isock = socket(AF_INET, SOCK_STREAM, 0); // 调用系统函数创建监听套接字
    if (isock == -1)
    {
        Log_Error(errno, "CSocket::Initialize()中socket()失败.");
        return false;
    }

    // 启用SO_REUSEADDR选项可以允许地址和端口的立即重用，解决占用问题
    int reuseaddr = 1;
    if (setsockopt(isock, SOL_SOCKET, SO_REUSEADDR, (const void *)&reuseaddr, sizeof(reuseaddr)) == -1)
    {
        Log_Error(errno, "CSocket::Initialize()中setsockopt(SO_REUSEADDR)失败.");
        close(isock); // 关闭套接字
        return false;
    }

    // 启用SO_REUSEPORT选项可以避免惊群问题
    int reuseport = 1;
    if (setsockopt(isock, SOL_SOCKET, SO_REUSEPORT, (const void *)&reuseport, sizeof(int)) == -1)
    {
        // 失败也没关系，惊群问题只影响效率
        Log_Error(errno, "CSocket::Initialize()中setsockopt(SO_REUSEPORT)失败");
    }

    // 设置该socket为非阻塞
    if (Setnonblocking(isock) == false)
    {
        Log_Error(errno, "CSocket::Initialize()中setnonblocking()失败.");
        close(isock);
        return false;
    }

    // 设置本服务器要监听的地址和端口
    iport = p_config->GetIntDefault("net_ListenPort", 11111); // 读取端口号
    serv_addr.sin_port = htons((in_port_t)iport);             // 转为大端序（in_port_t其实就是uint16_t）

    // 尝试绑定端口
    if (bind(isock, (struct sockaddr *)&serv_addr, sizeof(serv_addr)) == -1)
    {
        Log_Error(errno, "CSocket::Initialize()中bind()失败.");
        close(isock);
        return false;
    }

    // 尝试开启监听(LISTEN_BACKLOG 指定内核等待处理的连接队列最大长度，含半连接和全连接)
    if (listen(isock, LISTEN_BACKLOG) == -1)
    {
        Log_Error(errno, "CSocket::Initialize()中listen()失败.");
        close(isock);
        return false;
    }

    // 放入监听列表中
    lplistening_t p_listensocketitem = new listening_t; // 创建监听结构体
    memset(p_listensocketitem, 0, sizeof(listening_t));
    p_listensocketitem->port = iport;
    p_listensocketitem->fd = isock;
    Log_Info(0, "监听%d端口成功!", iport);
    m_ListenSocketList.push_back(p_listensocketitem);
    return true;
}

// 设置socket监听套接字为非阻塞模式(函数写法比较固定)
bool CSocket::Setnonblocking(int sockfd)
{
    int notBlock = 1; // 0:清除；1：设置
    // FIONBIO选项用于设置/清除非阻塞I/O标记
    if (ioctl(sockfd, FIONBIO, &notBlock) == -1)
    {
        return false;
    }
    return true;
}

// 关闭套接字的函数
void CSocket::Close_listening_sockets()
{
    // 端口只有一个
    close(m_ListenSocketList[0]->fd);
    Log_Info(0, "关闭监听端口%d!", m_ListenSocketList[0]->port);
}

// 将一个待发送消息放到消息队列中
void CSocket::MsgSend(char *psendbuf)
{
    CMemory *p_memory = CMemory::GetInstance();
    CLock lock(&m_sendMessageQueueMutex);

    // 防止消息队列过大给服务器带来的风险
    if (m_iSendMsgQueueCount > 50000)
    {
        // 过大就丢弃
        m_iDiscardSendPkgCount++;       // 记录丢弃的消息个数
        p_memory->FreeMemory(psendbuf); // 直接释放掉
        return;
    }

    // 检查消息是否是恶意的
    LPSTRUC_MSG_HEADER pMsgHeader = (LPSTRUC_MSG_HEADER)psendbuf;
    lpconnection_t p_Conn = pMsgHeader->pConn;
    // 检查这名用户在发送消息队列中积累的条数（过大可能是用户只发不收）
    if (p_Conn->iSendCount > 400)
    {
        // 发现此用户收消息太慢或只发不收，直接切断连接
        Log_Error(0, "CSocket::msgSend()中发现某用户%d积压了大量待发送数据包，切断与他的连接！", p_Conn->fd);
        m_iDiscardSendPkgCount++;
        p_memory->FreeMemory(psendbuf);
        ZdClosesocketProc(p_Conn); // 主动关闭一个连接处理函数
        return;
    }

    // 正常用户的发送处理
    ++p_Conn->iSendCount;
    m_MsgSendQueue.push_back(psendbuf);
    ++m_iSendMsgQueueCount;

    // 将信号量的值+1，唤醒发送消息处理线程
    if (sem_post(&m_semEventSendQueue) == -1) // 让ServerSendQueueThread()流程走下来干活
    {
        Log_Error(0, "CSocket::msgSend()中sem_post(&m_semEventSendQueue)失败.");
    }
}

// 主动关闭一个连接时的要做些善后的处理函数
void CSocket::ZdClosesocketProc(lpconnection_t p_Conn)
{
    // 如果开启了踢人时钟，就把这个队列里的去掉
    if (m_ifkickTimeCount == 1)
    {
        DeleteFromTimerQueue(p_Conn);
    }
    // 关闭套接字（会自动从epoll中删除）
    if (p_Conn->fd != -1)
    {
        close(p_Conn->fd);
        p_Conn->fd = -1;
    }
    // 自减发送缓冲区标志
    if (p_Conn->iThrowsendCount > 0)
    {
        --p_Conn->iThrowsendCount; // 其实这里置不置0都无所谓
    }
    // 放入到待回收队列
    InRecyConnectQueue(p_Conn);
}

// 检查是否发生Flood攻击
bool CSocket::TestFlood(lpconnection_t pConn)
{
    struct timeval sCurrTime; // 精确时间结构体（秒 + 微秒）
    uint64_t iCurrTime;       // 暂存当前时间（单位毫秒）

    gettimeofday(&sCurrTime, NULL); // 获取精确的当前时间
    // 转换为毫秒单位
    iCurrTime = (sCurrTime.tv_sec * 1000 + sCurrTime.tv_usec / 1000);
    // 如果两次收到包的时间 < 100毫秒
    if ((iCurrTime - pConn->FloodkickLastTime) < m_floodTimeInterval)
    {
        pConn->FloodAttackCount++; // 记录一次
        pConn->FloodkickLastTime = iCurrTime;
    }
    else
    {
        // 如果不频繁了就重置
        pConn->FloodAttackCount = 0;
        pConn->FloodkickLastTime = iCurrTime;
    }

    // 判断频繁次数是否超过了
    if (pConn->FloodAttackCount >= m_floodKickCount)
    {
        return true;
    }
    return false;
}

// 打印统计信息
void CSocket::PrintTDInfo()
{
    time_t currtime = time(NULL);
    if ((currtime - m_lastprintTime) > 10)
    {
        // 超过十秒打印一次
        int tmprmqc = g_threadpool.GetRecvMsgQueueCount(); // 收消息队列个数
        m_lastprintTime = currtime;
        int tmpoLUC = m_onlineUserCount;    // 在线用户数量
        int tmpsmqc = m_iSendMsgQueueCount; // 发送消息队列大小
        Log_Info(0, "------------------------------------begin--------------------------------------");
        Log_Info(0, "当前在线人数/总人数(%d/%d)。", tmpoLUC, m_worker_connections);
        Log_Info(0, "连接池中空闲连接/总连接/要释放的连接(%d/%d/%d)。", m_freeconnectionList.size(), m_connectionList.size(), m_recyconnectionList.size());
        Log_Info(0, "当前心跳超时队列大小(%d)。", m_timerQueuemap.size());
        Log_Info(0, "当前收消息队列/发消息队列大小分别为(%d/%d), 丢弃的待发送数据包数量为%d。", tmprmqc, tmpsmqc, m_iDiscardSendPkgCount);
        if (tmprmqc > 100000)
        {
            Log_Error(0, "接收队列条目数量过大(%d)，要考虑限速或者增加处理线程数量了！！！！！！", tmprmqc);
        }
        Log_Info(0, "-------------------------------------end---------------------------------------");
    }
}

// epoll功能初始化，在子进程进行
// 本函数被Worker_process_init函数调用
int CSocket::Epoll_init()
{
    // 很多内核版本不处理epoll_create的参数，只需要参数>0即可
    m_epollhandle = epoll_create(m_worker_connections); // 参数肯定>0
    if (m_epollhandle == -1)
    {
        Log_Error(errno, "CSocket::Epoll_init()中epoll_create()失败.");
        exit(2); // 这个问题很严重，直接退出程序，资源交由系统释放
    }

    // 创建套接字连接池
    Initconnection();

    // 遍历所有用于监听的套接字，绑定连接池中的一个对象
    std::vector<lplistening_t>::iterator pos;
    for (pos = m_ListenSocketList.begin(); pos != m_ListenSocketList.end(); ++pos)
    {
        lpconnection_t p_Conn = Get_connection((*pos)->fd);
        if (p_Conn == NULL)
        {
            // 说明此时连接池中的连接为空，这是致命问题
            Log_Error(errno, "CSocket::Epoll_init()中Get_connection()失败.");
            exit(2); // 这个问题很严重，直接退出程序，资源交由系统释放
        }
        p_Conn->listening = (*pos);  // 连接对象和监听对象关联，方便通过连接对象找监听对象
        (*pos)->connection = p_Conn; // 监听对象和连接对象关联，方便通过监听对象找连接对象

        // 对监听端口的读事件设置处理方法（处理新连接）
        p_Conn->rhandler = &CSocket::Event_accept;

        // 往监听套接字上增加监听事件，从而让其开始履行职责
        // 如果不加这个epoll注册部分，虽然端口能连上，但是不会触发epoll_wait()事件
        // 第一个参数是socket句柄
        // 第二个参数是事件类型，此处是EPOLL_CTL_ADD
        // 第三个参数是标志，此处EPOLLIN：可读，EPOLLRDHUP：TCP连接的远端关闭或者半关闭
        // 第四个参数是如果是事件增加类型就不需要，置0
        // 第五个参数是连接池中的连接
        if (Epoll_oper_event((*pos)->fd, EPOLL_CTL_ADD, EPOLLIN | EPOLLRDHUP, 0, p_Conn) == -1)
        {
            exit(2); // 这也是致命问题，直接退出程序，资源交由系统释放
        }
    }
    return 1;
}

// 控制操作epoll事件函数（如注册事件，修改事件、删除事件等）
// 第一个参数是socket句柄
// 第二个参数是事件类型，一般是EPOLL_CTL_ADD，EPOLL_CTL_MOD，EPOLL_CTL_DEL就是操作epoll红黑树的节点(增加，修改，删除)
// 第三个参数是标志，具体含义取决于eventtype
// 第四个参数是补充，用于补充flag标记的不足: 0增加;1去掉;2完全覆盖,eventtype是EPOLL_CTL_MOD时这个参数就有用
// 第五个参数是一个指针(其实是一个连接)将来epoll_wait时能取出来用
int CSocket::Epoll_oper_event(int fd, uint32_t eventtype, uint32_t flag, int bcaction, lpconnection_t pConn)
{
    struct epoll_event ev; // 用于承载epoll_wait中返回的事件
    memset(&ev, 0, sizeof(ev));

    // 如果是往epoll红黑树添加节点（监听的套接字）
    if (eventtype == EPOLL_CTL_ADD)
    {
        ev.events = flag;     // 既然是增加节点，则不管原来是啥标记
        pConn->events = flag; // 这个连接本身也记录这个标记
    }
    // 如果是修改红黑树中某个节点
    else if (eventtype == EPOLL_CTL_MOD)
    {
        ev.events = pConn->events; // 先取出标志
        if (bcaction == 0)
        { // 0表示增加
            ev.events |= flag;
        }
        // 1表示去掉
        else if (bcaction == 1)
        {
            ev.events &= ~flag;
        }
        // 否则表示直接覆盖
        else
        {
            ev.events = flag;
        }
        pConn->events = ev.events; // 保存新标志
    }
    // 否则就是删除节点
    else
    {
        // 但由于socket关闭会自动从红黑树移除，已经足够，目前没需求
        return 1;
    }

    ev.data.ptr = (void *)pConn; // 绑定连接指针，将来epoll_wait时能取出来用

    // 正式调用api进行操作
    if (epoll_ctl(m_epollhandle, eventtype, fd, &ev) == -1)
    {
        Log_Error(errno, "CSocket::Epoll_oper_event()中epoll_ctl(%d,%ud,%ud,%d)失败.", fd, eventtype, flag, bcaction);
        return -1;
    }
    return 1;
}

/**
 * Epoll轮询获取发生的事件（无论正不正常都要继续执行）
 * @param timer epoll_wait阻塞的时长(-1是一直阻塞)
 * @return 1正常，0不正常
 */
int CSocket::Epoll_process_events(int timer)
{
    // 事件会返回到m_events数组里面，最多返回配置的MAX_EVENTS个
    // 阻塞timer时长，除非：阻塞时间到达/阻塞期间收到事件(比如新用户连入)会立刻返回/调用时有事件也会立刻返回/如果来个信号，比如你用kill -1 pid测试
    int events = epoll_wait(m_epollhandle, m_events, MAX_EVENTS, timer);

    // 如果是有错误返回（-1）
    if (events == -1)
    {
        if (errno == EINTR)
        {
            // 如果是发送某个信号给本进程导致返回一个EINTR错误，错误码是4（被中断的系统调用）
            return 1; // 这个错误是正常的，正常返回
        }
        else
        {
            // 其他情况说明是真的出错了
            Log_Error(errno, "CSocket::Epoll_process_events()中epoll_wait失败!");
            return 0;
        }
    }
    // 如果是因为超时返回（0）
    if (events == 0)
    {
        if (timer != -1)
        {
            // 第一种情况：有超时时间，说明是时间到且没有事件发生，是正常的
            return 1; // 正常返回
        }
        // 第二种情况：明明设置无限等待，epoll_wait返回了但却没有任何事件发生，不正常
        Log_Info(0, "CSocket::Epoll_process_events()中epoll_wait()没超时却没返回任何事件!");
        return 0;
    }

    // 走到这里收到了发生的事件，进行处理
    lpconnection_t p_Conn;
    uint32_t revents; // 暂存事件类型
    for (int i = 0; i < events; ++i)
    {
        p_Conn = (lpconnection_t)(m_events[i].data.ptr); // 取出连接

        // 能走到这里，我们认为这些事件都没过期，就正常开始处理
        revents = m_events[i].events; // 取出事件类型

        // 如果是读事件
        if (revents & EPOLLIN)
        {
            (this->*(p_Conn->rhandler))(p_Conn); // 调用读函数
        }

        // 如果是写事件（注意对方关闭连接也触发这个）
        if (revents & EPOLLOUT)
        {
            // 客户端关闭，如果服务器端挂着一个写通知事件，则这里个条件是可能成立的
            if (revents & (EPOLLERR | EPOLLHUP | EPOLLRDHUP))
            {
                --p_Conn->iThrowsendCount; // 自减发送缓冲区标志
            }
            else
            {
                (this->*(p_Conn->whandler))(p_Conn);
            }
        }
    }
    return 1;
}

// 处理发送消息队列的线程函数
void *CSocket::ServerSendQueueThread(void *threadData)
{
    ThreadItem *pThread = static_cast<ThreadItem *>(threadData);
    CSocket *pSocketObj = pThread->_pThis;
    int err;
    std::list<char *>::iterator pos, posend;

    char *pMsgBuf;                 // 暂存发送缓冲区指针
    LPSTRUC_MSG_HEADER pMsgHeader; // 消息头指针
    LPCOMM_PKG_HEADER pPkgHeader;  // 包头指针
    lpconnection_t p_Conn;
    unsigned short itmp; // 暂时记录重新转为本机序的包长度
    ssize_t sendsize;    // 本次发送的数据长度（因为不保证一次发完）

    CMemory *p_memory = CMemory::GetInstance();

    // 循环直到程序退出
    while (g_stopEvent == 0)
    {
        if (sem_wait(&pSocketObj->m_semEventSendQueue) == -1)
        {
            // 注意如果是被某个信号中断而发生ETNTR信号返回的话，可以不用管
            if (errno != EINTR)
            {
                Log_Error(errno, "CSocket::ServerSendQueueThread()中sem_wait(&pSocketObj->m_semEventSendQueue)失败.");
            }
        }

        // 处理数据收发
        if (g_stopEvent != 0)
        {
            break; // 系统退出
        }

        // 如果发送消息队列有东西
        if (pSocketObj->m_iSendMsgQueueCount > 0)
        {
            err = pthread_mutex_lock(&pSocketObj->m_sendMessageQueueMutex);
            if (err != 0)
            {
                Log_Error(err, "CSocket::ServerSendQueueThread()中pthread_mutex_lock()失败，返回的错误码为%d!", err);
            }

            pos = pSocketObj->m_MsgSendQueue.begin();
            posend = pSocketObj->m_MsgSendQueue.end();

            while (pos != posend)
            {
                pMsgBuf = (*pos);                                                        // 容器中的消息是：消息头+包头+包体，但不发消息头给客户端
                pMsgHeader = (LPSTRUC_MSG_HEADER)pMsgBuf;                                // 得到消息头
                pPkgHeader = (LPCOMM_PKG_HEADER)(pMsgBuf + pSocketObj->m_iLenMsgHeader); // 指向包头
                p_Conn = pMsgHeader->pConn;                                              // 获取消息头存储的连接

                // 序列号对不上，说明是已经过期的了，直接删除
                // 这个序列化是每次TCP建立连接更新，防止发送混乱
                if (p_Conn->iCurrsequence != pMsgHeader->iCurrsequence)
                {
                    pos = pSocketObj->m_MsgSendQueue.erase(pos);
                    --pSocketObj->m_iSendMsgQueueCount; // 发送消息队列数量自减
                    p_memory->FreeMemory(pMsgBuf);
                    continue;
                }

                // 如果发送缓冲区已经满了，则需要通过epoll事件来驱动消息的继续发送
                // 所以这里不需要再进行发送
                if (p_Conn->iThrowsendCount > 0)
                {
                    pos++;
                    continue;
                }

                // 走到这说明可以发送这个消息
                p_Conn->psendMemPointer = pMsgBuf; // 发送完释放用的指针
                pos = pSocketObj->m_MsgSendQueue.erase(pos);
                --pSocketObj->m_iSendMsgQueueCount; // 发送队列长度自减
                --p_Conn->iSendCount;               // 记录的发送队列条数自减
                // 由于发送数据不一定能全部发送，记录发送到的位置，以便继续发送
                p_Conn->psendbuf = (char *)pPkgHeader; // 发送数据缓冲区的记录指针
                itmp = ntohs(pPkgHeader->pkgLen);
                p_Conn->isendlen = itmp; // 记录要发送的数据长度

                // 开始发送数据(调用发送函数)
                sendsize = pSocketObj->Sendproc(p_Conn, p_Conn->psendbuf, p_Conn->isendlen);
                if (sendsize > 0)
                { // >0说明发送没问题
                    if (sendsize == p_Conn->isendlen)
                    {                                                  // 说明一下发完了
                        p_Conn->iThrowsendCount = 0;                   // 标记发送缓冲区空闲
                        p_memory->FreeMemory(p_Conn->psendMemPointer); // 释放内存
                        p_Conn->psendMemPointer = NULL;
                    }
                    // 说明只发送了一部分
                    else
                    {
                        // 记录发送到了哪里，方便下次使用
                        p_Conn->psendbuf = p_Conn->psendbuf + sendsize;
                        p_Conn->isendlen = p_Conn->isendlen - sendsize;
                        // 标记一次发送缓冲区不空闲，需要通过epoll事件来驱动消息的继续发送
                        ++p_Conn->iThrowsendCount;
                        // 修改套接字的epoll事件设置，继续监听写事件
                        if (pSocketObj->Epoll_oper_event(p_Conn->fd, EPOLL_CTL_MOD, EPOLLOUT, 0, p_Conn) == -1)
                        {
                            // 这个问题会很麻烦，但是还是先不处理
                            Log_Error(errno, "CSocket::ServerSendQueueThread()Epoll_oper_event()失败.");
                        }
                    }
                    continue; // 继续下一个消息的发送
                }
                // 走到这里说明发送是出现了问题的，但是不影响，先认为发送完毕
                else if (sendsize == 0)
                {
                    p_memory->FreeMemory(p_Conn->psendMemPointer);
                    p_Conn->psendMemPointer = NULL;
                    p_Conn->iThrowsendCount = 0;
                    continue;
                }
                // 能走到这说明一个字节也没有发出去，发送缓冲区正好满了
                else if (sendsize == -1)
                {
                    // 标记一次发送缓冲区不空闲，需要通过epoll事件来驱动消息的继续发送
                    ++p_Conn->iThrowsendCount;
                    // 修改套接字的epoll事件设置，继续监听写事件
                    if (pSocketObj->Epoll_oper_event(p_Conn->fd, EPOLL_CTL_MOD, EPOLLOUT, 0, p_Conn) == -1)
                    {
                        // 这个问题会很麻烦，但是还是先不处理
                        Log_Error(errno, "CSocket::ServerSendQueueThread()Epoll_oper_event2()失败.");
                    }
                    continue;
                }
                // 能走到这一般是返回-2，说明对端断开了，等待recv()来处理
                else
                {
                    // 直接舍弃消息，等待被回收
                    p_memory->FreeMemory(p_Conn->psendMemPointer);
                    p_Conn->psendMemPointer = NULL;
                    p_Conn->iThrowsendCount = 0;
                    continue;
                }
            }

            err = pthread_mutex_unlock(&pSocketObj->m_sendMessageQueueMutex);
            if (err != 0)
            {
                Log_Error(err, "CSocket::ServerSendQueueThread()pthread_mutex_unlock()失败，返回的错误码为%d!", err);
            }
        }
    }

    return (void *)0;
}